package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

import static com.heima.common.constants.HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
import static com.heima.common.constants.HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;

@Slf4j
@Configuration
public class KafkaSteam {

//    @Bean
//    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
//        //接收消息
//        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
//        //聚合流式处理
//        stream.map((key, value) -> {
//                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
//                    //重置消息的key:1234343434   和  value: likes:1
//                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
//                })
//                //按照文章id进行聚合
//                .groupBy((key, value) -> key)
//                //时间窗口
//                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
//                /**
//                 * 自行的完成聚合的计算
//                 */
//                .aggregate(new Initializer<String>() {
//                    /**
//                     * 初始方法，返回值是消息的value
//                     * @return
//                     */
//                    @Override
//                    public String apply() {
//                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
//                    }
//                    /**
//                     * 真正的聚合操作，返回值是消息的value
//                     */
//                }, new Aggregator<String, String, String>() {
//                    @Override
//                    public String apply(String key, String value, String aggValue) {
//                        if (StringUtils.isBlank(value)) {
//                            return aggValue;
//                        }
//                        String[] aggAry = aggValue.split(",");
//                        int col = 0, com = 0, lik = 0, vie = 0;
//                        for (String agg : aggAry) {
//                            String[] split = agg.split(":");
//                            /**
//                             * 获得初始值，也是时间窗口内计算之后的值
//                             */
//                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
//                                case COLLECTION:
//                                    col = Integer.parseInt(split[1]);
//                                    break;
//                                case COMMENT:
//                                    com = Integer.parseInt(split[1]);
//                                    break;
//                                case LIKES:
//                                    lik = Integer.parseInt(split[1]);
//                                    break;
//                                case VIEWS:
//                                    vie = Integer.parseInt(split[1]);
//                                    break;
//                            }
//                        }
//                        /**
//                         * 累加操作
//                         */
//                        String[] valAry = value.split(":");
//                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
//                            case COLLECTION:
//                                col += Integer.parseInt(valAry[1]);
//                                break;
//                            case COMMENT:
//                                com += Integer.parseInt(valAry[1]);
//                                break;
//                            case LIKES:
//                                lik += Integer.parseInt(valAry[1]);
//                                break;
//                            case VIEWS:
//                                vie += Integer.parseInt(valAry[1]);
//                                break;
//                        }
//
//                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
//                        System.out.println("文章的id:" + key);
//                        System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
//                        return formatStr;
//                    }
//                }, Materialized.as("hot-atricle-stream-count-001"))
//                .toStream()
//                .map((key, value) -> {
//                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value));
//                })
//                //发送消息
//                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
//
//        return stream;
//
//
//    }
//
//    /**
//     * 格式化消息的value数据
//     *
//     * @param articleId
//     * @param value
//     * @return
//     */
//    public String formatObj(String articleId, String value) {
//        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
//        mess.setArticleId(Long.valueOf(articleId));
//        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
//        String[] valAry = value.split(",");
//        for (String val : valAry) {
//            String[] split = val.split(":");
//            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
//                case COLLECTION:
//                    mess.setCollect(Integer.parseInt(split[1]));
//                    break;
//                case COMMENT:
//                    mess.setComment(Integer.parseInt(split[1]));
//                    break;
//                case LIKES:
//                    mess.setLike(Integer.parseInt(split[1]));
//                    break;
//                case VIEWS:
//                    mess.setView(Integer.parseInt(split[1]));
//                    break;
//            }
//        }
//        log.info("聚合消息处理之后的结果为:{}", JSON.toJSONString(mess));
//        return JSON.toJSONString(mess);
//
//    }

    @Bean
    public KStream<String, String> kStream(StreamsBuilder builder) {

        KStream<String, String> stream = builder.stream(HOT_ARTICLE_SCORE_TOPIC);


        stream.map((key, value) -> {
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
                }).groupBy((Key, value) -> Key).windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new Initializer<String>() {
                               @Override
                               public String apply() {
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                           }, new Aggregator<String, String, String>() {
                               @Override
                               public String apply(String key, String value, String aggValues) {

                                   if (StringUtils.isBlank(value)) {
                                       return aggValues;
                                   }
                                   System.out.println(aggValues);
                                   int col = 0, com = 0, lik = 0, vie = 0;
                                   String[] aggValue = aggValues.split(",");
                                   for (String agg : aggValue) {

                                       String[] split = agg.split(":");
                                       switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                           case COLLECTION:
                                               col = Integer.parseInt(split[1]);
                                               break;
                                           case COMMENT:
                                               com = Integer.parseInt(split[1]);
                                               break;
                                           case LIKES:
                                               int i = Integer.parseInt(split[1]);
                                               System.out.println("i:" + i);
                                               lik = i;
                                               break;
                                           case VIEWS:
                                               vie = Integer.parseInt(split[1]);
                                               break;
                                       }
                                   }
                                   String[] values = value.split(":");
                                   switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])) {
                                       case COLLECTION:
                                           col += Integer.parseInt(values[1]);
                                           break;
                                       case COMMENT:
                                           com += Integer.parseInt(values[1]);
                                           break;
                                       case LIKES:
                                           lik += Integer.parseInt(values[1]);
                                           break;
                                       case VIEWS:
                                           vie += Integer.parseInt(values[1]);
                                           break;
                                   }
                                   String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                                   System.err.println(formatStr);
                                   System.out.println("articleId:" + key);
                                   return formatStr;
                               }
                           },
                        Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key, value) -> new KeyValue<>(key.key(), formatObj(key.key(), value)))
                .to(HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;


    }

    private String formatObj(String id, String value) {
        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
        articleVisitStreamMess.setArticleId(Long.valueOf(id));
        String[] values = value.split(",");
        for (String val : values) {
            String[] split = val.split(":");

            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    articleVisitStreamMess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    articleVisitStreamMess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    articleVisitStreamMess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    articleVisitStreamMess.setView(Integer.parseInt(split[1]));
                    break;
            }

        }
        return JSON.toJSONString(articleVisitStreamMess);
    }

}